package com.settlement.system.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.settlement.system.common.base.IBaseEnum;
import com.settlement.system.common.enums.DeleteEnum;
import com.settlement.system.common.enums.SendErrorRecordStatusEnum;
import com.settlement.system.common.util.StringUtils;
import com.settlement.system.mapper.KafkaSendErrorRecordMapper;
import com.settlement.system.model.entity.KafkaSendErrorRecord;
import com.settlement.system.model.form.KafkaSendErrorRecordForm;
import com.settlement.system.model.query.KafkaSendErrorRecordQuery;
import com.settlement.system.model.vo.KafkaSendErrorRecordImportVo;
import com.settlement.system.service.KafkaSendErrorRecordService;
import lombok.RequiredArgsConstructor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

/**
 * <p>
 * KAFKA发送失败记录表 服务实现类
 * </p>
 *
 * @author huangrensen
 * @since 2024-09-09
 */
@Service
@RequiredArgsConstructor
public class KafkaSendErrorRecordServiceImpl extends ServiceImpl<KafkaSendErrorRecordMapper, KafkaSendErrorRecord> implements KafkaSendErrorRecordService {

    private final Logger logger = LoggerFactory.getLogger(KafkaSendErrorRecordServiceImpl.class);

    private final KafkaTemplate<String, String> kafkaTemplate;

    @Override
    public Page<KafkaSendErrorRecord> getPage(KafkaSendErrorRecordQuery queryParams) {
        // 查询参数
        int pageNum = queryParams.getPageNum();
        int pageSize = queryParams.getPageSize();
        // 查询数据
        Page<KafkaSendErrorRecord> page = new Page<>(pageNum, pageSize);
        Page<KafkaSendErrorRecord> mapperPage = this.baseMapper.getPage(page, queryParams);
        if (mapperPage.getRecords().size() > 0) {
            for (KafkaSendErrorRecord record : mapperPage.getRecords()) {
                record.setStatusName(IBaseEnum.getLabelByValue(record.getStatus(), SendErrorRecordStatusEnum.class));
            }
        }
        return mapperPage;
    }

    @Override
    public boolean saveKafkaSendErrorRecord(KafkaSendErrorRecordForm kafkaSendErrorRecordForm) {
        KafkaSendErrorRecord kafkaSendErrorRecord = new KafkaSendErrorRecord();
        BeanUtils.copyProperties(kafkaSendErrorRecordForm, kafkaSendErrorRecord);
        kafkaSendErrorRecord.setCreateTime(null);
        return this.saveOrUpdate(kafkaSendErrorRecord);
    }

    @Override
    public KafkaSendErrorRecordForm getKafkaSendErrorRecordFormData(Long id) {
        KafkaSendErrorRecord kafkaSendErrorRecord = this.getById(id);
        KafkaSendErrorRecordForm kafkaSendErrorRecordForm = new KafkaSendErrorRecordForm();
        BeanUtils.copyProperties(kafkaSendErrorRecord, kafkaSendErrorRecordForm);
        return kafkaSendErrorRecordForm;
    }

    @Override
    public void checkAndSave(KafkaSendErrorRecord kafkaSendErrorRecord) {
        logger.info("根据【{}】【{}】查找是否存在记录...", kafkaSendErrorRecord.getQueueName(), kafkaSendErrorRecord.getUuid());
        List<KafkaSendErrorRecord> records = this.list(new LambdaQueryWrapper<KafkaSendErrorRecord>()
                .eq(KafkaSendErrorRecord::getQueueName, kafkaSendErrorRecord.getQueueName())
                .eq(KafkaSendErrorRecord::getUuid, kafkaSendErrorRecord.getUuid()));
        if (records.size() > 0) {
            logger.info("存在记录，更新操作...");
            KafkaSendErrorRecord first = records.get(0);
            first.setStatus(SendErrorRecordStatusEnum.ERROR.getValue());
            first.setRemark(kafkaSendErrorRecord.getRemark());
            first.setUpdateTime(LocalDateTime.now());
            first.cutString();
            this.updateById(kafkaSendErrorRecord);
        } else {
            logger.info("不存在记录，新增操作...");
            kafkaSendErrorRecord.setUpdateTime(LocalDateTime.now());
            kafkaSendErrorRecord.cutString();
            this.save(kafkaSendErrorRecord);
        }
    }

    @Override
    public void reSendAll() {
        KafkaSendErrorRecordQuery queryParams = new KafkaSendErrorRecordQuery();
        queryParams.setPageNum(1);
        queryParams.setPageSize(300);
        Page<KafkaSendErrorRecord> page = this.getPage(queryParams);
        while (page.getRecords().size() > 0) {
            logger.info("查询第{}页，共{}页...", queryParams.getPageNum(), page.getPages());
            for (KafkaSendErrorRecord record : page.getRecords()) {
                record.setStatus(SendErrorRecordStatusEnum.RUNNING.getValue());
                record.setUpdateTime(LocalDateTime.now());
            }
            this.updateBatchById(page.getRecords());
            for (KafkaSendErrorRecord record : page.getRecords()) {
                CompletableFuture<SendResult<String, String>> completableFuture = kafkaTemplate.send(record.getQueueName(), record.getDataText());
                completableFuture.whenComplete((result, ex) -> {
                    if (Objects.nonNull(ex)) {
                        record.setRemark(ex.getMessage());
                        this.checkAndSave(record);
                    }else{
                        // 标志删除
                        this.deleteKafkaSendErrorRecord(record.getId().toString());
                    }
                });
            }
            // 下一页
            try {
                Thread.sleep(3000);
            } catch (Exception e) {
            }
            queryParams.setPageNum(queryParams.getPageNum() + 1);
            logger.info("查询第{}页，共{}页...", queryParams.getPageNum(), page.getPages());
            page = this.getPage(queryParams);
        }
    }

    @Override
    public boolean deleteKafkaSendErrorRecord(String ids) {
        Assert.isTrue(StringUtils.hasText(ids), "删除的数据为空");
        List<Long> idArray = Arrays.stream(ids.split(",")).map(Long::parseLong).collect(Collectors.toList());
        List<KafkaSendErrorRecord> errorRecords = this.getBaseMapper().selectBatchIds(idArray);
        for (KafkaSendErrorRecord errorRecord : errorRecords) {
            logger.info("标志为删除数据：{}", errorRecord);
            errorRecord.setDeleted(DeleteEnum.YES.getValue());
        }
        return this.updateBatchById(errorRecords);
    }

    @Override
    public void importData(List<KafkaSendErrorRecordImportVo> voList) {
        List<KafkaSendErrorRecord> kafkaSendErrorRecordList = new ArrayList<>();
        for (KafkaSendErrorRecordImportVo importVo : voList) {
            KafkaSendErrorRecord kafkaSendErrorRecord = new KafkaSendErrorRecord();
            BeanUtils.copyProperties(importVo, kafkaSendErrorRecord);
            kafkaSendErrorRecordList.add(kafkaSendErrorRecord);
        }
        this.saveOrUpdateBatch(kafkaSendErrorRecordList);
    }

    @Override
    public List<KafkaSendErrorRecordImportVo> findByIds(String idsStr) {
        if (!StringUtils.hasText(idsStr)) {
            return new ArrayList<>();
        }
        List<Long> ids = Arrays.stream(idsStr.split(",")).map(Long::parseLong).collect(Collectors.toList());
        List<KafkaSendErrorRecord> kafkaSendErrorRecordList = this.getBaseMapper().selectBatchIds(ids);
        List<KafkaSendErrorRecordImportVo> voList = new ArrayList<>();
        for (KafkaSendErrorRecord kafkaSendErrorRecord : kafkaSendErrorRecordList) {
            KafkaSendErrorRecordImportVo vo = new KafkaSendErrorRecordImportVo();
            BeanUtils.copyProperties(kafkaSendErrorRecord, vo);
            voList.add(vo);
        }
        return voList;
    }
}
